package com.ck.ips.schedule;

import com.ck.common.model.MonitorDataFile;
import com.ck.common.model.MonitorDataProgram;
import com.ck.core.constant.DateFormatConstant;
import com.ck.core.constant.ElasticsearchConstant;
import com.ck.core.lock.LockConstant;
import com.ck.core.lock.RedisDistributeLock;
import com.ck.core.lock.RedisLock;
import com.ck.core.util.DateUtils;
import com.ck.ips.core.config.CustomProperties;
import com.ck.ips.service.impl.ElasticsearchServiceImpl;
import com.ck.ips.service.program.MonitorDataFileService;
import com.ck.ips.service.program.MonitorDataProgramService;
import com.google.common.collect.Lists;
import io.searchbox.core.SearchResult;
import io.searchbox.core.search.aggregation.MetricAggregation;
import io.searchbox.core.search.aggregation.SumAggregation;
import io.searchbox.core.search.aggregation.TermsAggregation;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.MatchAllQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.aggregations.metrics.sum.SumAggregationBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.text.ParseException;
import java.util.List;

/**
 * 清洗监播数据定时任务
 *
 * @author dxy
 * @date 2019/5/8 10:20
 */
@Component
public class MonitorDataJobs {
	private Logger logger = LoggerFactory.getLogger(MonitorDataJobs.class);

	@Autowired
	private ElasticsearchServiceImpl elasticsearchService;
	@Autowired
	private MonitorDataFileService monitorDataFileService;
	@Autowired
	private MonitorDataProgramService monitorDataProgramService;
	@Autowired
	private RedisDistributeLock redisDistributeLock;
	@Autowired
	private CustomProperties customProperties;

	/**
	 * 清洗监播数据（素材）
	 */
	@Scheduled(cron = "0 30 3 * * ?")
	public void saveMonitorDataFile() {
		logger.debug("清洗监播数据saveMonitorDataFile...");
		// job加锁，避免在分布式环境多个应用同时执行任务
		try ( RedisLock lock = redisDistributeLock.getLock(LockConstant.MONITOR_FILE_LOCK_PREFIX + this.getClass().getName(),
				LockConstant.DEFAULT_MONITOR_TIMEOUT_MILLS, LockConstant.MAX_MONITOR_RETRY_TIMES, LockConstant.RETRY_INTERVAL_TIME_MILLIS) ) {
			if (null != lock) {
				logger.debug("Starting saveMonitorDataFile schedule...");

				//获取昨天的日期
				String date = DateUtils.getPastDay(DateFormatConstant.YESTERDAY, DateFormatConstant.DATE_YEAR_MONTH_DAY);
				//索引名称
				String indexName = customProperties.getBusinessType() + ElasticsearchConstant.INDEX_MONITOR_DATA_FILE;
				//索引类型
				String indexType = "";

				try {
					Long startTime = DateUtils.stringTimeToLongTimeStamp(date + " " + DateFormatConstant.MIN_HOUR_MINUTE_SECOND_MILLISECOND, DateFormatConstant.YEAR_MONTH_DAY_HMS_SSS);
					//计算一天中的最大日期
					Long endTime = DateUtils.stringTimeToLongTimeStamp(date + " " + DateFormatConstant.MAX_HOUR_MINUTE_SECOND_MILLISECOND, DateFormatConstant.YEAR_MONTH_DAY_HMS_SSS);
					//素材时间查询条件
					RangeQueryBuilder queryBuilder = QueryBuilders.rangeQuery(ElasticsearchConstant.FILE_PLAY_START_TIME).from(startTime).to(endTime).format(ElasticsearchConstant.EPOCH_MILLIS);

					//查询索引中的所有数据
					MatchAllQueryBuilder matchAllQueryBuilder = QueryBuilders.matchAllQuery();
					//多条件查询
					BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

					boolQueryBuilder.must(matchAllQueryBuilder).must(queryBuilder);

					//按终端编码分组
					TermsAggregationBuilder deviceCodeAggregationBuilder = AggregationBuilders.terms(ElasticsearchConstant.GROUP_BY_DEVICE_CODE).field(ElasticsearchConstant.DEVICE_CODE_KEYWORD).size(Integer.MAX_VALUE);

					//按programVersion分组
					TermsAggregationBuilder programVersionAggregationBuilder = AggregationBuilders.terms(ElasticsearchConstant.GROUP_BY_PROGRAM_VERSION).field(ElasticsearchConstant.PROGRAM_VERSION_KEYWORD).size(Integer.MAX_VALUE);

					//按fileNameURL分组
					TermsAggregationBuilder fileNameURLAggregationBuilder = AggregationBuilders.terms(ElasticsearchConstant.GROUP_BY_FILE_NAME_URL).field(ElasticsearchConstant.FILE_NAME_URL_KEYWORD).size(Integer.MAX_VALUE);
					//计算累计播放时长
					SumAggregationBuilder playDurationSumAggregationBuilder = AggregationBuilders.sum(ElasticsearchConstant.PLAY_DURATION_SUM).field(ElasticsearchConstant.PLAY_DURATION);

					fileNameURLAggregationBuilder.subAggregation(playDurationSumAggregationBuilder);
					programVersionAggregationBuilder.subAggregation(fileNameURLAggregationBuilder);
					deviceCodeAggregationBuilder.subAggregation(programVersionAggregationBuilder);

					//从ES查询数据
					SearchResult searchResult = elasticsearchService.doSearch(indexName, indexType, null, boolQueryBuilder, deviceCodeAggregationBuilder);
					String errorMessage = searchResult.getErrorMessage();
					if (StringUtils.isNotBlank(errorMessage)) {
						logger.error("MonitorDataJobs saveMonitorDataFile errorMessage ---> " + errorMessage);
					} else {
						MetricAggregation aggregations = searchResult.getAggregations();
						TermsAggregation deviceCodeTermsAggregation = aggregations.getTermsAggregation(ElasticsearchConstant.GROUP_BY_DEVICE_CODE);
						//获取deviceCode分组的Buckets
						List<TermsAggregation.Entry> deviceCodeBuckets = deviceCodeTermsAggregation.getBuckets();

						//监播数据（素材）列表
						List<MonitorDataFile> monitorDataFileList = Lists.newArrayList();

						for (TermsAggregation.Entry deviceCodeEntry : deviceCodeBuckets) {
							//终端编码
							String deviceCode = deviceCodeEntry.getKey();

							TermsAggregation programVersionTermsAggregation = deviceCodeEntry.getTermsAggregation(ElasticsearchConstant.GROUP_BY_PROGRAM_VERSION);
							//节目版本分组的Buckets
							List<TermsAggregation.Entry> programVersionBuckets = programVersionTermsAggregation.getBuckets();

							for (TermsAggregation.Entry programVersionEntry : programVersionBuckets) {
								//节目版本
								String programVersion = programVersionEntry.getKey();

								//获取fileNameURL分组的Buckets
								TermsAggregation fileNameURLTermsAggregation = programVersionEntry.getTermsAggregation(ElasticsearchConstant.GROUP_BY_FILE_NAME_URL);
								List<TermsAggregation.Entry> fileNameURLBuckets = fileNameURLTermsAggregation.getBuckets();

								for (TermsAggregation.Entry fileNameURLEntry : fileNameURLBuckets) {
									String fileNameURL = fileNameURLEntry.getKey();
									//素材名称
									String fileName = "";
									//素材URL
									String fileURL = "";
									String[] fileNameURLArr = fileNameURL.split(ElasticsearchConstant.SEPARATOR);
									if (fileNameURLArr != null && fileNameURLArr.length > 0){
										fileName = fileNameURLArr[0];
										fileURL = fileNameURLArr[1];
									}
									//获取播放次数(fileNameURL分组就是播放次数)
									Long playNum = fileNameURLEntry.getCount();
									if (playNum == null) {
										playNum = 0L;
									}
									//获取累计播放时长
									SumAggregation playDurationSumAggregation = fileNameURLEntry.getSumAggregation(ElasticsearchConstant.PLAY_DURATION_SUM);
									Double palyDurationSum = playDurationSumAggregation.getSum();
									if (palyDurationSum == null) {
										palyDurationSum = 0D;
									}

									MonitorDataFile monitorDataFile = new MonitorDataFile();
									monitorDataFile.setPlayDate(date);
									monitorDataFile.setProgramVersion(programVersion);
									monitorDataFile.setDeviceCode(deviceCode);
									monitorDataFile.setFileName(fileName);
									monitorDataFile.setFileUrl(fileURL);
									monitorDataFile.setPlayNum(playNum.intValue());
									monitorDataFile.setPlayDuration(palyDurationSum.intValue());
									monitorDataFile.setAddTime(System.currentTimeMillis());
									monitorDataFileList.add(monitorDataFile);
								}

							}
						}
						//存入数据库
						if (CollectionUtils.isNotEmpty(monitorDataFileList)) {
							monitorDataFileService.insertBatch(monitorDataFileList);
							logger.info("MonitorDataJobs saveMonitorDataFile ---> 保存成功");
						}
					}
				} catch (ParseException e) {
					logger.error("MonitorDataJobs saveMonitorDataFile ParseException ---> " + e.getMessage());
				} catch (IOException e) {
					logger.error("MonitorDataJobs saveMonitorDataFile IOException  ---> " + e.getMessage());
				}

			}
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
		}


	}

	/**
	 * 清洗监播数据（节目）
	 */
	@Scheduled(cron = "0 0 4 * * ?")
	public void saveMonitorDataProgram() {
		// job加锁，避免在分布式环境多个应用同时执行任务
		logger.debug("清洗监播数据（节目）Starting saveMonitorDataProgram schedule...");
		try ( RedisLock lock = redisDistributeLock.getLock(LockConstant.MONITOR_PROGRAM_LOCK_PREFIX + this.getClass().getName(),
				LockConstant.DEFAULT_MONITOR_TIMEOUT_MILLS, LockConstant.MAX_MONITOR_RETRY_TIMES, LockConstant.RETRY_INTERVAL_TIME_MILLIS) ) {
			if (null != lock) {
				logger.debug("Starting saveMonitorDataProgram schedule...");

				//获取昨天的日期
				String date = DateUtils.getPastDay(DateFormatConstant.YESTERDAY, DateFormatConstant.DATE_YEAR_MONTH_DAY);
				//索引名称
				String indexName = customProperties.getBusinessType() +ElasticsearchConstant.INDEX_MONITOR_DATA_PROGRAM;
				//索引类型
				String indexType = "";

				try {
					Long startTime = DateUtils.stringTimeToLongTimeStamp(date + " " + DateFormatConstant.MIN_HOUR_MINUTE_SECOND_MILLISECOND, DateFormatConstant.YEAR_MONTH_DAY_HMS_SSS);
					//计算一天中的最大日期
					Long endTime = DateUtils.stringTimeToLongTimeStamp(date + " " + DateFormatConstant.MAX_HOUR_MINUTE_SECOND_MILLISECOND, DateFormatConstant.YEAR_MONTH_DAY_HMS_SSS);
					//节目时间查询条件
					RangeQueryBuilder queryBuilder = QueryBuilders.rangeQuery(ElasticsearchConstant.PROGRAM_PLAY_START_TIME).from(startTime).to(endTime).format(ElasticsearchConstant.EPOCH_MILLIS);

					//查询索引中的所有数据
					MatchAllQueryBuilder matchAllQueryBuilder = QueryBuilders.matchAllQuery();
					//多条件查询
					BoolQueryBuilder boolQueryBuilder = QueryBuilders.boolQuery();

					boolQueryBuilder.must(matchAllQueryBuilder).must(queryBuilder);

					//按programVersion分组
					TermsAggregationBuilder programVersionAggregationBuilder = AggregationBuilders.terms(ElasticsearchConstant.GROUP_BY_PROGRAM_VERSION).field(ElasticsearchConstant.PROGRAM_VERSION_KEYWORD).size(Integer.MAX_VALUE);

					//计算节目累计播放时长
					SumAggregationBuilder playDurationSumAggregationBuilder = AggregationBuilders.sum(ElasticsearchConstant.PLAY_DURATION_SUM).field(ElasticsearchConstant.PLAY_DURATION);

					programVersionAggregationBuilder.subAggregation(playDurationSumAggregationBuilder);

					//从ES查询数据
					SearchResult searchResult = elasticsearchService.doSearch(indexName, indexType, null, boolQueryBuilder, programVersionAggregationBuilder);
					String errorMessage = searchResult.getErrorMessage();
					if (StringUtils.isNotBlank(errorMessage)) {
						logger.error("MonitorDataJobs saveMonitorDataProgram errorMessage ---> " + errorMessage);
					} else {
						MetricAggregation aggregations = searchResult.getAggregations();
						TermsAggregation programVersionTermsAggregation = aggregations.getTermsAggregation(ElasticsearchConstant.GROUP_BY_PROGRAM_VERSION);
						//获取programVersionDeviceCode分组的Buckets
						List<TermsAggregation.Entry> programVersionBuckets = programVersionTermsAggregation.getBuckets();

						//监播数据（节目）列表
						List<MonitorDataProgram> monitorDataProgramList = Lists.newArrayList();

						for (TermsAggregation.Entry programVersionEntry : programVersionBuckets) {
							String programVersion = programVersionEntry.getKey();
							//获取播放次数
							Long playNum = programVersionEntry.getCount();
							if (playNum == null) {
								playNum = 0L;
							}
							//获取累计播放时长
							SumAggregation playDurationSumAggregation = programVersionEntry.getSumAggregation(ElasticsearchConstant.PLAY_DURATION_SUM);
							Double palyDurationSum = playDurationSumAggregation.getSum();
							if (palyDurationSum == null) {
								palyDurationSum = 0D;
							}

							MonitorDataProgram monitorDataProgram = new MonitorDataProgram();
							monitorDataProgram.setPlayDate(date);
							monitorDataProgram.setProgramVersion(programVersion);
							monitorDataProgram.setPlayNum(playNum.intValue());
							monitorDataProgram.setPlayDuration(palyDurationSum.longValue());
							monitorDataProgram.setAddTime(System.currentTimeMillis());
							monitorDataProgramList.add(monitorDataProgram);
						}
						//存入数据库
						if (CollectionUtils.isNotEmpty(monitorDataProgramList)) {
							monitorDataProgramService.insertBatch(monitorDataProgramList);
							logger.info("MonitorDataJobs saveMonitorDataProgram ---> 保存成功");
						}
					}
				} catch (ParseException e) {
					logger.error("MonitorDataJobs saveMonitorDataProgram ParseException ---> " + e.getMessage());
				} catch (IOException e) {
					logger.error("MonitorDataJobs saveMonitorDataProgram IOException  ---> " + e.getMessage());
				}

			}
		} catch (Exception e) {
			logger.error(e.getMessage(),e);
		}
	}
}
